package cn.tedu.micro.demo.emqx.mqtt;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.nio.charset.StandardCharsets;

/**
 * 即将交给MqttClient连接使用的回调对象
 */
@Slf4j
public class MqttMyCallback implements MqttCallbackExtended {
    private MqttClient client;
    private String groupName;
    //创建钩子函数 构造方法
    public MqttMyCallback(MqttClient client,String groupName){
        this.client=client;
        this.groupName=groupName;
    }
    /**
     * 回调时机: 客户端mqttClient成功建立与emqx服务端连接关系时调用
     * boolean reconnect: true表示本次连接是重连,false表示第一次连接
     * String serverUri: 实际连接的地址 tcp://192.168.8.100:1883
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        if(reconnect){
            //重连
            log.info("重新连接成功,broker地址:{}",serverURI);
        }else{
            log.info("首次连接成功,broker地址:{}",serverURI);
        }
        //让当前客户端MqttClient订阅需要消费的主题名称
        try{
            //订阅topic01 共享订阅 当前多个消费者属于同一组 争抢消息
            client.subscribe("$share/g1/topic01");
            //订阅topic02 普通订阅 多个消费者同时消费消息
            client.subscribe("topic02");
            //订阅主题topic03
            client.subscribe("$share/"+groupName+"/topic03");
        }catch (Exception e){
            log.error("订阅主题失败",e);
        }
    }
    /**
     * 回调时机: mqttClient连接断开时调用
     * Throwable cause: 连接断开原因
     * 如果设置了setAutomaticReconnect(true).自动重连,否则这个代码 手动处理重新连接
     */
    @Override
    public void connectionLost(Throwable cause) {
        log.info("连接断开",cause);
    }
    /**
     * 回调时机: 当前回调对象所属的MqttClient如果订阅了某些主题,如果主题中出现消息
     * 才会调用这个方法 客户端可以同时订阅多个主题
     * String topic: 消息来源主题
     * MqttMessage message: 消息对象(payLoad,qos retain)
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        String msg=new String(message.getPayload(), StandardCharsets.UTF_8);
        log.info("收到消息,来源:{},消息内容:{}",topic,msg);
    }
    /**
     * 回调时机: mqttClient发送publish消息的时候,如果成功了,收到emqx反馈ack=true就会调用
     * 这个方法,但是只有在QOS>0的时候才有效 因为  QOS=0 客户端不关心成功失败,不接收反馈
     * IMqttDeliveryToken token: 包含消息ID等上下文信息 原始消息就在token里
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("消息发送成功,消息标识:{}",token.getMessageId());
    }
}
